package mylab.io.store.mqtt.client;

import com.google.common.base.Preconditions;
import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAckReturnCode;
import com.hivemq.client.mqtt.mqtt3.message.subscribe.suback.Mqtt3SubAckReturnCode;

import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import cn.hutool.core.util.EnumUtil;
import mylab.base.foundation.lang.Callback;
import mylab.io.base.BaseDas;
import mylab.io.base.context.DefaultDC;
import mylab.io.store.mqtt.common.MqttMsg;
import mylab.io.store.mqtt.common.MqttQoS;
import mylab.io.store.mqtt.property.MqttClientProperties;
import mylab.utils.common.ByteUtil;
import mylab.utils.common.ExceptionUtil;

public class HivemqMqttClientDas extends BaseDas<DefaultDC<MqttClientProperties, MqttMsg>, MqttClientProperties, MqttMsg>
        implements mylab.io.store.mqtt.client.MqttClient {

    Mqtt3AsyncClient client;

    public HivemqMqttClientDas(DefaultDC<MqttClientProperties, MqttMsg> context) {
        super(context);

        client = com.hivemq.client.mqtt.MqttClient.builder()
                .identifier(UUID.randomUUID().toString())
                .serverAddress(properties.getBrokerAddress())
                .useMqttVersion3()
                .build().toAsync();
    }

    @Override
    protected void startFunc(Callback<Void> callback) {
        client.connect().whenCompleteAsync((ack, throwable) -> {
            if (throwable != null) {
                Callback.dealOnException(callback, throwable);
            } else {
                boolean isConnected = Objects.equals(Mqtt3ConnAckReturnCode.SUCCESS, ack.getReturnCode());
                if (isConnected) {
                    Callback.dealOnSuccess(callback);
                } else {
                    Callback.dealOnFailure(callback, ack.getReturnCode().toString());
                }
            }
        });

    }

    @Override
    protected void stopFunc(Callback<Void> callback) {
        client.disconnect().whenCompleteAsync((unused, throwable) -> {
            if (throwable != null) {
                Callback.dealOnException(callback, throwable);
            } else {
                Callback.dealOnSuccess(callback);
            }
        });
    }

    @Override
    protected void asyncSendFunc(MqttMsg msg, Callback<Void> callback) {
        publish(msg.getTopic(), msg.getQos(), msg.getPayload(), callback);
    }

    @Override
    public MqttMsg syncSend(MqttMsg msg, long timeout, TimeUnit unit) throws Exception {
        ExceptionUtil.throwException("mqttClient is not supported to syncSend");
        return null;
    }

    @Override
    public void publish(String topic, MqttQoS qoS, byte[] payload, Callback<Void> callback) {
        checkClient();

        client.publishWith()
                .topic(topic)
                .qos(convertQos(qoS))
                .payload(payload)
                .send()
                .whenCompleteAsync((publish, throwable) -> {
                    if (throwable != null) {
                        Callback.dealOnException(callback, throwable);
                    } else {
                        Callback.dealOnSuccess(callback);
                    }
                });

    }

    @Override
    public void subscribe(String topic, MqttQoS qoS, Callback<MqttMsg> callback) {
        checkClient();

        client.subscribeWith()
                .topicFilter(topic)
                .qos(convertQos(qoS))
                .callback(publish -> {
                    MqttQoS qos = convertQos(publish.getQos());
                    String content = ByteUtil.bytesToString(publish.getPayloadAsBytes());
                    MqttMsg res = new MqttMsg(topic, content, qos);

                    getContext().onReceived(res);
                    Callback.dealOnSuccess(callback, res);
                })
                .send()
                .whenCompleteAsync((ack, throwable) -> {
                    if (throwable != null) {
                        Callback.dealOnException(callback, throwable);
                    } else {
                        //全部订阅成功才为成功，只要有一条订阅失败即为失败
                        boolean hasFailure = ack.getReturnCodes().stream().anyMatch(code -> Objects.equals(Mqtt3SubAckReturnCode.FAILURE, code));
                        if (hasFailure) {
                            Callback.dealOnFailure(callback, "订阅失败");
                        }
                    }
                });
    }

    @Override
    public void unsubscribe(String topic, Callback<Void> callback) {
        checkClient();

        client.unsubscribeWith()
                .topicFilter(topic)
                .send()
                .whenCompleteAsync((unused, throwable) -> {
                    if (throwable != null) {
                        Callback.dealOnException(callback, throwable);
                    } else {
                        Callback.dealOnSuccess(callback);
                    }
                });
    }

    void checkClient() {
        Preconditions.checkState(client.getState().isConnected(), "mqtt client is not connected");
    }

    static com.hivemq.client.mqtt.datatypes.MqttQos convertQos(MqttQoS qos) {
        return EnumUtil.fromString(com.hivemq.client.mqtt.datatypes.MqttQos.class, qos.name());
    }

    static MqttQoS convertQos(com.hivemq.client.mqtt.datatypes.MqttQos qos) {
        return EnumUtil.fromString(MqttQoS.class, qos.name());
    }
}
